package project

import java.io

import dmputils.PeiZhiFile
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/03/29.
  */
object MtFenXi {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    .setMaster("local[*]")
    .setAppName("媒体分析")
    .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    val sc: SparkContext = new SparkContext(conf)

    val sqlc: SQLContext = new SQLContext(sc)

    import sqlc.implicits._

    val app_dictfile: RDD[String] = sc.textFile(PeiZhiFile.config.getString("app_dictpath"))

    val splitApp_dictData: RDD[Array[String]] = app_dictfile.map(_.split("\t"))

    //splitApp_dictData.collect().foreach(println)

    val GlApp_dictData: RDD[(String, String)] = splitApp_dictData.map(t=>{
      val tuple = if (t.length >= 5) {
        (t(4),t(1))
      } else {
        (null, null)
      }
      tuple
    })

    val GlApp_dictDatabroadcast: Broadcast[RDD[(String, String)]] = sc.broadcast(GlApp_dictData)

    val GlApp_dictmap: Map[String, String] = GlApp_dictDatabroadcast.value.map(t=>(t._1,t._2)).collect().toMap


    val basedata: DataFrame = sqlc.read.parquet(PeiZhiFile.config.getString("inputpath"))

    val basedataSx: RDD[(String, Int, Int, Int, Int, Int, Int, Int, Double, Double)] = basedata.map(row => {

      val newappname: String = if (row.getAs[String]("appname").equals("其他") || row.getAs[String]("appname").equals("未知")) {

        val appname1: String = GlApp_dictmap.getOrElse(row.getAs[String]("appid"),row.getAs[String]("appid"))
        appname1
      } else {
        row.getAs[String]("appname")
      }

      //数据请求方式（1请求、2展示、3点击）
      val requestmode: Int = row.getAs[Int]("requestmode")
      //流程节点（1：请求量 kpi 2：有效请求 3：广告请求
      val processnode: Int = row.getAs[Int]("processnode")
      //有效标识（有效指可以正常计费的）(0：无效 1：有效
      val iseffective: Int = row.getAs[Int]("iseffective")
      //是否收费（0：未收费 1：已收费）
      val isbilling: Int = row.getAs[Int]("isbilling")
      //是否 rtb
      val isbid: Int = row.getAs[Int]("isbid")
      //是否竞价成功
      val iswin: Int = row.getAs[Int]("iswin")
      // 广告 id
      val adorderid: Int = row.getAs[Int]("adorderid")
      //rtb 竞价成功价格 每次消费
      val winprice: Double = row.getAs[Double]("winprice")
      //转换后的广告消费
      val adpayment: Double = row.getAs[Double]("adpayment")

      (newappname, requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment)

    })
    val pdData: RDD[(String, Int, Int, Int, Int, Int, Int, Int, Double, Double)] = basedataSx.map(t => {
      //原始请求数
      val Ysrequest = if (t._2 == 1 && t._3 >= 1) 1 else 0
      //有效请求数
      var Yxrequest = if (t._2 == 1 && t._3 >= 2) 1 else 0
      //广告请求数
      var Ggrequest = if (t._2 == 1 && t._3 == 3) 1 else 0
      //参与竞价数
      var Cyjinjia = if (t._4 == 1 && t._5 == 1 && t._6 == 1 && t._8 != 0) 1 else 0
      //竞价成功数
      var Jjsuccess = if (t._4 == 1 && t._5 == 1 && t._7 == 1) 1 else 0
      //展示数
      var Zssome = if (t._2 == 2 && t._4 == 1) 1 else 0
      //点击数
      var Djsome = if (t._2 == 3 && t._4 == 1) 1 else 0
      //DSP广告消费
      var Xfmoney = if (t._4 == 1 && t._5 == 1 && t._7 == 1) t._9/1000 else 0
      //DSP广告成本
      var Cbmoney = if (t._4 == 1 && t._5 == 1 && t._7 == 1) t._10/1000 else 0

      (t._1, Ysrequest, Yxrequest, Ggrequest, Cyjinjia, Jjsuccess, Zssome, Djsome, Xfmoney, Cbmoney)
    })
    val pdDataFrame = pdData.toDF("应用名称","原始请求","有效请求","广告请求","参与竞价数","竞价成功数","展示量","点击量","广告成本","广告消费")

    pdDataFrame.registerTempTable("t_MtFenxi")


    //sqlc.sql("select * from t_MtFenxi").show()
    val sql = sqlc.sql(
      """
        |select 应用名称  应用名称,
        |sum (原始请求) 原始请求,
        |sum (有效请求) 有效请求,
        |sum (广告请求) 广告请求,
        |sum (参与竞价数) 参与竞价数,
        |sum (竞价成功数) 竞价成功数,
        |sum (展示量) 展示量,
        |sum (点击量) 点击量,
        |sum (广告成本) 广告成本,
        |sum (广告消费) 广告消费
        |from t_MtFenxi
        |group by 应用名称
      """.stripMargin)
    val map = sql.map(x=>x)
    //sql.write.parquet()

    map.coalesce(4).saveAsTextFile("G:\\TestData\\DMPData\\MtFenxi2")

    //sql.coalesce(4).write.mode(SaveMode.Append).save("G:\\TestData\\DMPData\\MtFenxi")










    sc.stop()
  }


}
